package edu.scut.dap.service;

import edu.scut.dap.entity.TaskState;
import edu.scut.dap.util.StartRequest;
import edu.scut.dap.util.StateRequest;
import edu.scut.dap.util.StateResponse;
import edu.scut.dap.util.StopRequest;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by Alex on 2016/12/3.
 */
public class TaskCMServiceImpl implements TaskCMService.Iface{
    Logger logger=Logger.getLogger(TaskCMService.class);
    List<TaskState> taskStateList= new ArrayList<>();

    public int startTask(StartRequest request) throws TException {
        System.out.println(request);
        List<String> startArgs=new ArrayList<String>();
        startArgs.add("yarn");
        startArgs.add("jar");
        startArgs.add(request.getOperatorName());
        startArgs.add(request.getInput());
        if (request.isSetAnotherInput()){
            startArgs.add(request.getAnotherInput());
        }
        startArgs.add(request.getOutput());
        if (request.getTaskArgsSize()!=0){
            startArgs.addAll(request.getTaskArgs());
        }
        System.out.println(startArgs);

        try {
            Process process=Runtime.getRuntime().exec(startArgs.toArray(new String[startArgs.size()]));
            TaskState taskState=new TaskState().setTaskName(request.getTaskName()).setProcess(process).setTimestampNow();
            taskState.startUpdateMessage();
            int taskid;
            synchronized (taskStateList){
                taskid=taskStateList.size();
                taskStateList.add(taskState);
            }
            logger.info("Task id "+taskid+" is running");
            return taskid;
        } catch (IOException e) {
            e.printStackTrace();
            logger.error("Fail to start task, args of which are: "+ request.toString());
            return -1;
        }
    }

    public int stopTask(StopRequest request) throws TException {
        return -1;
    }

    public StateResponse taskState(StateRequest request) throws TException {
        try{
            TaskState taskState=taskStateList.get(request.getTaskid());
            StateResponse stateResponse=new StateResponse(taskState.getTaskName(),taskState.getExitCode(),taskState.getRunTime());
            return stateResponse;
        }catch (IndexOutOfBoundsException e){
            logger.error("Fail to get task state, task id "+request.getTaskid()+" doesn't exsist");
            return null;
        }
    }

    public String getOutputMessage(StateRequest request) throws TException {
        try{
            TaskState taskState=taskStateList.get(request.getTaskid());
            String outputMessage=taskState.getOutputMessage();
            System.out.println(outputMessage);
            return outputMessage;
        }catch (IndexOutOfBoundsException e){
            logger.error("Fail to get output message, task id "+request.getTaskid()+" doesn't exsist");
            return null;
        }
    }

    public String getErrorMessage(StateRequest request) throws TException {
        try{
            TaskState taskState=taskStateList.get(request.getTaskid());
            String errorMessage=taskState.getErrorMessage();
            return errorMessage;
        }catch (IndexOutOfBoundsException e){
            logger.error("Fail to get error message, task id "+request.getTaskid()+" doesn't exsist");
            return null;
        }
    }

    @Override
    public List<StateResponse> allTaskState() throws TException {
        List<StateResponse> responseList= new ArrayList<>();
        try{
            synchronized (taskStateList){
                for (int i = 0; i < taskStateList.size(); i++) {
                    TaskState taskState=taskStateList.get(i);
                    StateResponse stateResponse=new StateResponse(taskState.getTaskName(),taskState.getExitCode(),taskState.getRunTime());
                    responseList.add(stateResponse);
                }
            }
            return responseList;
        }catch (IndexOutOfBoundsException e){
            logger.error("Fail to get all task state");
            return null;
        }
    }
}
